//OMP c++ wrapper
//$Id: easy_omp.h 139 2009-09-02 16:21:19Z Oleg.Bulychov $
#pragma once
#ifndef EASY_OMP_H
#define EASY_OMP_H

#include <deque>
#include <omp.h>
#include "src/heo/include/reflection.h"
#include "src/heo/include/transmit.h"
#include "src/heo/include/sys_api.h"

//_______________________________________________________________________________
//_______________________________________________________________________________
namespace omp
{
//_______________________________________________________________________________
namespace hidden
{
//_______________________________________________________________________________
//TODO test for speed:
                              //boost::circular_buffer<byte_t>;
                              //std::vector<byte_t>
template < class M, class P = std::deque<byte_t> >
class thread_data_buffer
{
    DISALLOW_COPY_AND_ASSIGN(thread_data_buffer);
public:
    typedef M                               TMutex;
    typedef P                               TPipe;
    typedef typename TMutex::scoped_lock    TLock;
//.............................................................................
    TPipe*          pipe_;
    int             num_threads_;
    volatile int    flag_;
    TMutex          mutex_;
//.............................................................................
    thread_data_buffer():
        pipe_(NULL),
        num_threads_(0),
        flag_(0),
        mutex_()
    {}
//.............................................................................
    ~thread_data_buffer()
    {
        delete[] pipe_;
    }
//.............................................................................
    void init(int num_threads)
    {
        num_threads_ = num_threads;
        pipe_ = new TPipe[num_threads];
    }
//.............................................................................
    void clear_Data()
    {
        for (int i = 0; i < num_threads_; ++i)
        {
            pipe_[i].clear();
        }
        flag_ = 0;
    }
};
//_______________________________________________________________________________
template<class T>
class thread_buffer_list
{
    DISALLOW_COPY_AND_ASSIGN(thread_buffer_list);

    typedef typename T::TLock               TLock;
    typedef typename T::TPipe               TPipe;

    T*  list_;
    int num_threads_;
public:
    thread_buffer_list(): list_(NULL), num_threads_(0)
    {}
//.............................................................................
    ~thread_buffer_list()
    {
        delete[] list_;
    }
//.............................................................................
    void init(int num_threads)
    {//:NOTE: not thread safe!
        num_threads_ = num_threads;
        list_ = new T[num_threads];
        for (int i = 0; i < num_threads; ++i)
            list_[i].init(num_threads);
    }
//.............................................................................
    void clear_Data(int rank)
    {
        ASSERT_(rank < num_threads_);

        TLock lock(list_[rank].mutex_);
        list_[rank].clear_Data();
    }
//.............................................................................
    void isend(int rank, int dst, byte_t const* pv, size_t cb)
    {
        ASSERT_(rank < num_threads_);
        ASSERT_(dst < num_threads_);

        {
            TLock lock(list_[dst].mutex_);
            {
                TPipe& pipe = list_[dst].pipe_[rank];
                ////size_t old_size = pipe.size();
                ////pipe.resize(pipe.size() + cb);
                ////memcpy(&pipe[old_size], pv, cb);
                pipe.insert(pipe.end(), pv, pv + cb);
                ++list_[dst].flag_;
            }
        }
    }
//.............................................................................
    bool is_Data_Available(int rank, int src)
    {
        ASSERT_(rank < num_threads_);
        ASSERT_(src < num_threads_);

        if (!list_[rank].flag_)
            return false;
        TLock lock;
        if (!lock.try_acquire(list_[rank].mutex_))
            return false;
        return !list_[rank].pipe_[src].empty();
    }
//.............................................................................
    bool irecv(int rank, int src, byte_t* pv, size_t cb)
    {
        ASSERT_(rank < num_threads_);
        ASSERT_(src < num_threads_);
        if (!list_[rank].flag_)
            return false;
        TLock lock;
        if (!lock.try_acquire(list_[rank].mutex_))
            return false;
        TPipe& pipe = list_[rank].pipe_[src];
        if (pipe.empty())
            return false;
        std::copy(pipe.begin(), pipe.begin() + cb, pv);
        ////memcpy(pv, &pipe[0], cb);
        //stdext::unchecked_copy(pipe.begin(), pipe.begin() + cb, pv);
        pipe.erase(pipe.begin(), pipe.begin() + cb);
        --list_[rank].flag_;
        return true;
    }
//.............................................................................
    void recv(int rank, int src, byte_t* pv, size_t cb)
    {
        while(!irecv(rank, src, pv, cb))
        {
            sys_api::Yield_Thread();
        }
    }
};
//_______________________________________________________________________________
//_______________________________________________________________________________
template<class M>
class omp_visitor:
    public meta::visitor<M>
{
public:
    typedef meta::visitor< M > TVisitor;
    using TVisitor::self;
//.............................................................................
    template<typename T, typename D>
    void visit(T& t, char const* szName, D const& d)
    {
        visit2(t, meta::int2type<meta::is_integral_type<T>::result>());
    }
    template<typename T>
    void visit2(T& t, meta::int2type<0>)
    {
        transmit_handler<T, M>::visit(t, *self());
    }
    template<typename T>
    void visit2(T& t, meta::int2type<1>)
    {
        self()->invoke(&t, 1);
    }
//.............................................................................
    template<typename T, typename D>
    void visit_Array(T* pt, size_t N, char const* szName, D const& d)
    {
        visit_Array2(pt, N, szName, d, meta::int2type<meta::is_integral_type<T>::result>());
    }
    template<typename T, typename D>
    void visit_Array2(T* pt, size_t N, char const* szName, D const& d, meta::int2type<0> it)
    {
        TVisitor::visit_Array(pt, N, szName, d);
    }
    template<typename T, typename D>
    void visit_Array2(T* pt, size_t N, char const* szName, D const& d, meta::int2type<1>)
    {
        self()->invoke(pt, N);
    }
//.............................................................................
};
//_______________________________________________________________________________
}//hidden
//_______________________________________________________________________________
class mutex
{
    DISALLOW_COPY_AND_ASSIGN(mutex);
    omp_lock_t handle_;
public:
    mutex()
    {
        omp_init_lock(&handle_);
    }
    ~mutex()
    {
        omp_destroy_lock(&handle_);
    }

    class scoped_lock
    {
        DISALLOW_COPY_AND_ASSIGN(scoped_lock);
        mutex* m_;
    public:
        scoped_lock(): m_(NULL)
        {}

        explicit scoped_lock(mutex& m): m_(&m)
        {
            omp_set_lock(&m.handle_);
        }

        ~scoped_lock()
        {
            if (m_)
                omp_unset_lock(&m_->handle_);
        }

        bool try_acquire(mutex& m)
        {
            ASSERT_(!m_);
            if (omp_test_lock(&m.handle_))
            {
                m_ = &m;
                return true;
            }
            return false;
        }
    };
};
//_______________________________________________________________________________
/*
template<typename T = int>
class atomic
{
    volatile T value_;
    DISALLOW_COPY_AND_ASSIGN(atomic);
public:
    atomic(T v = 0): value_(v)
    {}

    atomic<T>& operator ++()
    {
#pragma omp atomic
        ++value_;
        return *this;
    }

    atomic<T>& operator --()
    {
#pragma omp atomic
        --value_;
        return *this;
    }

    operator volatile T()
    {
        return value_;
    }
};
*/
//_______________________________________________________________________________
//_______________________________________________________________________________
//_______________________________________________________________________________
class communicator
{
    typedef hidden::thread_buffer_list< hidden::thread_data_buffer<mutex> > TBuffList;
    TBuffList&                      itc_;
    int                             num_threads_;
    int                             rank_;
public:
    communicator(TBuffList& itc, int num_threads, int rank):
        itc_(itc), num_threads_(num_threads), rank_(rank)
    {}
//.............................................................................
    template<class T>
    static void run(int num_threads, T& t)
    {//:NOTE: not thread safe!
        TBuffList itc;
        itc.init(num_threads);
        volatile int i = 0;
#pragma omp parallel num_threads(num_threads)
        {
            int rank;
            {
#pragma omp critical
                rank = i++;
            }
            communicator tg(itc, num_threads, rank);
            t(tg);
        }
    }
//.............................................................................
    int size() const
    {
        return num_threads_;
    }
//.............................................................................
    int rank() const
    {
        return rank_;
    }
//.............................................................................
    void barrier()
    {
#pragma omp barrier
    }
//.............................................................................
    void isend(int dst, byte_t const* pv, size_t cb)
    {
        return itc_.isend(rank_, dst, pv, cb);
    }
//.............................................................................
    bool is_Data_Available(int src)
    {
        return itc_.is_Data_Available(rank_, src);
    }
//.............................................................................
    void recv(int src, byte_t* pv, size_t cb)
    {
        return itc_.recv(rank_, src, pv, cb);
    }
//.............................................................................
    void clear_Data()
    {
        itc_.clear_Data(rank_);
    }
};
//_______________________________________________________________________________
class isend:
    public hidden::omp_visitor<isend>
{
    int dst_;
    communicator& comm_;
public:
    enum { transmit_tag = SEND_TAG };
public:
    isend(int dest, communicator& comm): dst_(dest),
        comm_(comm)
    {}
//.............................................................................
    template<typename T>
    isend& operator <<(T& t)
    {
        visit_T(t);
        return *this;
    }
//.............................................................................
    template<typename T>
    void invoke(T* pt, size_t count)
    {
        COMPILETIME_CHECK(meta::is_integral_type<T>::result, bad_value);
        comm_.isend(dst_, (byte_t*)pt, count * sizeof(T));
    }
};
//_______________________________________________________________________________
class recv:
    public hidden::omp_visitor<recv>
{
    int src_;
    communicator& comm_;
public:
    enum { transmit_tag = RECV_TAG };
public:
    recv(int src, communicator& comm): src_(src),
        comm_(comm)
    {}
//.............................................................................
    template<typename T>
    recv& operator >>(T& t)
    {
        visit_T(t);
        return *this;
    }
//.............................................................................
    template<typename T>
    void invoke(T* pt, size_t count)
    {
        COMPILETIME_CHECK(meta::is_integral_type<T>::result, bad_value);
        comm_.recv(src_, (byte_t*)pt, count * sizeof(T));
    }
//.............................................................................
    bool is_Data_Available()
    {
        return comm_.is_Data_Available(src_);
    }
};
//_______________________________________________________________________________
}//omp
//__________________________________________________________________________________
//__________________________________________________________________________________
#endif
//EOF!
